package com.lsx143.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.lsx143.realtime.bean.TableProcess;
import com.lsx143.realtime.common.Constants;
import com.lsx143.realtime.util.JdbcUtil;
import com.lsx143.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * 自定义PhoenixSink,用于通过Phoenix皮肤保存数据到HBase
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    //phoenix连接对象
    private Connection phoenixCon;
    //保存是否第一次建表的状态
    private ValueState<Boolean> createdTableState;
    private Jedis jedis;

    /**
     * 初始化调用一次,进行Phoenix连接JDBC到hbase
     *
     * @param parameters 一些JDBC连接参数
     * @throws Exception 报错
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        //1.JDBC连接到HBase
        phoenixCon = JdbcUtil.getJdbcConnection(Constants.PHOENIX_DRIVER, Constants.PHOENIX_URL);
        //2.获取redis
        jedis = RedisUtil.getRedisClient();
        //3.保存第一次建表的状态
        createdTableState = getRuntimeContext()
                .getState(new ValueStateDescriptor<>("createdTableState", Boolean.class));
    }

    /**
     * 关闭jdbc连接
     *
     * @throws Exception 异常
     */
    @Override
    public void close() throws Exception {
        if (phoenixCon != null) {
            phoenixCon.close();
        }
    }

    /**
     * 执行的业务逻辑
     *
     * @param value 输入流
     * @param ctx   上下文对象
     * @throws Exception 一次
     */
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,
                       Context ctx) throws Exception {
        //1.执行建表语句,如果之前没有建表,则自动进行创建
        checkTable(value);
        //2.写入数据到hbase
        writeDataToHBase(value);
        //3.更新数据到redis
        updateDataToRedis(value);
    }

    /**
     * 更新redis中保存的DIM维度层数据
     * 必须在写入DIM层数据到HBase之后,同步update的操作类型的数据更新到Redis中
     * 否则在redis中读到的可能就是旧的数据
     *
     * @param value 输入流
     */
    private void updateDataToRedis(Tuple2<JSONObject, TableProcess> value) {
        JSONObject jsonOBJ = value.f0;
        TableProcess tb = value.f1;

        String key = tb.getSinkTable() + ":" + jsonOBJ.getLong("id");
        //判断操作是否为update和redis中是否已经存在key
        //是的话更新redis中的值
        if ("update".equals(tb.getOperateType()) && jedis.exists(key)) {
            System.out.println("更新redis数据");
            JSONObject result = new JSONObject();
            for (Map.Entry<String, Object> entries : jsonOBJ.entrySet()) {
                String upperKey = entries.getKey().toUpperCase();
                Object v = entries.getValue();
                result.put(upperKey, v);
            }
            jedis.setex(key, Constants.REDIS_EXPIRE_TIME, result.toJSONString());
        }
    }

    /**
     * 判断是否第一次写入数据,第一次写入则进行建表
     *
     * @param value 待写入hbase的数据
     * @throws SQLException 异常
     */
    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        //是否第一次建表
        if (createdTableState.value() == null) {
            TableProcess tp = value.f1;
            System.out.println("表:" + tp.getSinkTable() + " 为首次建表");
            //建表语句: create table if not exists table_name(id varchar, name varchar,
            // sku_id varchar, constraint pk primary key(id,name)) SALT_BUCKETS = 3

            //1.对字段语句进行拼接
            String columns = tp.getSinkColumns().concat(",").replaceAll(",", " varchar,");
            String cleanColumns = columns.substring(0, columns.length() - 1);
            //2.建表语句拼接
            StringBuffer sql = new StringBuffer();
            sql
                    .append("create table if not exists ")
                    .append(tp.getSinkTable())
                    .append("(")
                    .append(cleanColumns)
                    //这个为高级写法
                    //.append(tp.getSink_columns().replaceAll("([^,]+)", "$1 varchar"))
                    .append(", constraint pk primary key(")
                    .append(tp.getSinkPk() == null ? "id" : tp.getSinkPk())
                    .append("))")
                    .append(tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
            System.out.println("【建表】执行建表语句:" + sql);
            //3.执行sql
            PreparedStatement ps = phoenixCon.prepareStatement(String.valueOf(sql));
            ps.execute();
            //4.提交事务
            phoenixCon.commit();
            ps.close();
            //5.更新状态
            createdTableState.update(true);
        } else {
            System.out.println("表:" + value.f1.getSinkTable() + " 已存在,无需再次建表");
        }
    }

    /**
     * 写出数据到HBase
     *
     * @param tuple 输入的数据流
     * @throws SQLException SQL异常
     */
    @SuppressWarnings("all")
    private void writeDataToHBase(Tuple2<JSONObject, TableProcess> tuple) throws SQLException {
        JSONObject data = tuple.f0;
        TableProcess tp = tuple.f1;

        //1.编写sql语句
        //upsert into table_name(id,name) values(?,?)
        StringBuffer sql = new StringBuffer();
        sql
                .append("upsert into ")
                .append(tp.getSinkTable())
                .append("(")
                .append(tp.getSinkColumns())
                .append(") values(")
                .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
                .append(")");

        //2.执行sql
        PreparedStatement ps = phoenixCon.prepareStatement(String.valueOf(sql));
        //  2.1)替换通配符
        String[] columnName = tp.getSinkColumns().split(",");
        for (int i = 0; i < columnName.length; i++) {
            Object objValue = data.get(columnName[i]);
            String value = objValue == null ? null : objValue.toString();
            ps.setString(i + 1, value);
        }
        //  2.2)执行
        ps.execute();
        phoenixCon.commit();
        ps.close();
    }
}

